package com.codejiwei.flink.practice;

import com.codejiwei.flink.entity.UserBehavior;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author jiwei
 * @description 统计网站总浏览量(PV)
 * @date 2023/5/23 8:40
 */
public class Flink_Project_Practice_PV2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        env.readTextFile("E:\\workspace\\ss-flink\\src\\main\\resources\\data\\UserBehavior.csv")
                .map(line -> {
                    String[] fields = line.split(",");
                    return new UserBehavior(
                            Long.valueOf(fields[0]),
                            Long.valueOf(fields[1]),
                            Integer.valueOf(fields[2]),
                            fields[3],
                            Long.valueOf(fields[4])
                    );
                })
                .filter(user -> "pv".equals(user.getBehavior()))
                .keyBy(UserBehavior::getBehavior)
                .process(new KeyedProcessFunction<String, UserBehavior, Tuple2<String, Long>>() {
                    long count = 0;
                    @Override
                    public void processElement(UserBehavior userBehavior, Context context, Collector<Tuple2<String, Long>> collector) throws Exception {
                        count++;
                        collector.collect(Tuple2.of(userBehavior.getBehavior(), count));
                    }
                })
                .print();

        env.execute();
    }
}
